package org.niit.sql

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

//使用Spark SQL连接MySQL
object SparkSQL_2 {
  def main(args: Array[String]): Unit = {
    //准备环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL")
    val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
    val sparkContext: SparkContext = spark.sparkContext
    sparkContext.setLogLevel("ERROR")


    //读取数据库
    val dataFrame: DataFrame = spark.read
      .format("jdbc") //连接mysql 是用 jdbc 连接oracle  odbc
      .option("url", "jdbc:mysql://node1:3306/BD1")
      .option("driver", "com.mysql.jdbc.Driver") //数据库驱动
      .option("user", "root") //数据库用户名
      .option("password", "Niit@123") //数据库密码
      .option("dbtable", "movie") //数据库中的表
      .load() //加载。将该表中的数据封装到DataFrame中
    dataFrame.show()


    //写入数据库
    dataFrame.write
      .format("jdbc") //连接mysql 是用 jdbc 连接oracle  odbc
      .option("url", "jdbc:mysql://node1:3306/BD1")
      .option("driver", "com.mysql.jdbc.Driver") //数据库驱动
      .option("user", "root") //数据库用户名
      .option("password", "Niit@123") //数据库密码
      .option("dbtable", "movie2") //数据库中的表
      .mode(SaveMode.Append)//当该表不存在的时候，会自动创建
      .save() //加载。将该表中的数据封装到DataFrame中

    spark.close()
  }
}
